package com.xes.rabbitmq.core.service.impl;

import com.xes.rabbitmq.core.domain.tables.pojos.TbStudent;
import com.xes.rabbitmq.core.domain.tables.records.TbMessageRecord;
import com.xes.rabbitmq.core.enums.MessageStatusEunm;
import com.xes.rabbitmq.core.service.SenderStudentService;
import lombok.extern.slf4j.Slf4j;
import org.jooq.DSLContext;
import org.jooq.UpdateConditionStep;
import org.jooq.UpdateSetFirstStep;
import org.jooq.UpdateSetMoreStep;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import static com.xes.rabbitmq.core.domain.tables.TbMessage.TB_MESSAGE;


@Slf4j
@Service
public class SenderStudentServiceImpl implements SenderStudentService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private DSLContext dsl;

    /**
     * 发送消息
     *
     * @param student
     */
    public void send(TbStudent student, String messageId) {
        rabbitTemplate.setConfirmCallback(confirmCallback);

        CorrelationData correlationData = new CorrelationData(messageId);
        rabbitTemplate.convertAndSend(
                "student-exchange",
                "student-#",
                student,
                correlationData
        );
    }


    /**
     * 消息回调方法
     */
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            log.info("消息确认，消息ID={}", correlationData.getId());
            UpdateSetFirstStep<TbMessageRecord> table = dsl.update(TB_MESSAGE);
            UpdateConditionStep<TbMessageRecord> where = null;
            if (b) {
                UpdateSetMoreStep<TbMessageRecord> columns = table.set(TB_MESSAGE.STATUS, MessageStatusEunm.SUCCESS.getValue());
                where = columns.where(TB_MESSAGE.ID.eq(correlationData.getId()));
            } else {
                // ack false 有可能消息队列满了，有可能MQ集群同步复制异常
                UpdateSetMoreStep<TbMessageRecord> columns = table.set(TB_MESSAGE.STATUS, MessageStatusEunm.FAIL.getValue());
                where = columns.where(TB_MESSAGE.ID.eq(correlationData.getId())
                        .and(TB_MESSAGE.STATUS.ne(MessageStatusEunm.SUCCESS.getValue())));
            }
            int rows = where.execute();
            log.info("消息执行完毕，共影响{}行", rows);
        }
    };
}
